package dongshi.daddy.byzantine.pbft;

import java.io.*;
import java.net.InetSocketAddress;
import java.net.ServerSocket;
import java.net.Socket;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;

public class PbftTest {
    private static CountDownLatch allNodeReady = new CountDownLatch(5);
    private static CountDownLatch notExists = new CountDownLatch(1);
    private static CountDownLatch prePrepareCdl = new CountDownLatch(3);
    private static CountDownLatch prepareCdl = new CountDownLatch(6);
//    private static CountDownLatch commitCdl = new CountDownLatch(9);

    private static String SUQIN_NAME = "苏秦";
    private static int SUQIN_PORT = 6666;
    private static String ZHAO_NAME = "leader赵";
    private static int ZHAO_PORT = 6667;
    private static String WEI_NAME = "魏";
    private static int WEI_PORT = 6668;
    private static String HAN_NAME = "韩";
    private static int HAN_PORT = 6669;
    private static String CHU_NAME = "叛徒楚";
    private static int CHU_PORT = 6670;
    private static String ATTACK_MSG = "attack";
    private static String RETREAT_MSG = "retreat";

    // 客户端请求
    private static CountDownLatch request = new CountDownLatch(1);

    public static void main(String[] args) throws InterruptedException {
        // 客户端
        Thread suqin = new Thread(() -> {
            try {
                String whoAmI = SUQIN_NAME;
                ServerSocket serverSocket = new ServerSocket(SUQIN_PORT);
                allNodeReady.countDown();
                new Thread(() -> {
//                    int counter = 0;
                    List<String> commitMsgList = new ArrayList<>();
                    while (commitMsgList.size() <= 2) {
                        // counter++;
                        try {
                            String msg = receiveMsg(serverSocket);
                            commitMsgList.add(msg);
                            System.out.println(whoAmI + " 收到提交成功消息：" + msg);
                            // 达成共识啦✿✿ヽ(°▽°)ノ✿
                            if (commitMsgList.size() > 2) {
                                System.out.println("达成共识啦✿✿ヽ(°▽°)ノ✿");
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();

                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                // 苏秦发进攻消息给赵
                sendMsg(ATTACK_MSG, ZHAO_PORT);
                request.await();


            } catch (Exception e) {
                e.printStackTrace();
            }
        }, SUQIN_NAME);
        suqin.start();

        // leader
        Thread zhao = new Thread(() -> {
            try {
                String whoAmI = ZHAO_NAME;
                ServerSocket serverSocket = new ServerSocket(ZHAO_PORT);
                allNodeReady.countDown();
                new Thread(() -> {
//                    int counter = 0;
                    // 客户端请求的消息
                    String requestMsg = "";
                    Set<String> allReceivedMsgSet = new HashSet<>();
                    List<String> prepareMsgList = new ArrayList<>();
                    List<String> commitPhaseMsgList = new ArrayList<>();
                    while (true) {
//                        counter++;
                        try {
                            String msg = receiveMsg(serverSocket);
                            // 如果请求客户端消息为空，则说明是请求消息
                            if (requestMsg == "") {
                                requestMsg = msg;
                                allReceivedMsgSet.add(msg);
                                System.out.println("请求(leader)" + whoAmI + "收到消息: " + msg);
                                // 预准备阶段，将消息发送给魏楚韩
                                try {
                                    sendMsg(msg, CHU_PORT);
                                    sendMsg(msg, WEI_PORT);
                                    sendMsg(msg, HAN_PORT);
                                } catch (Exception e) {
                                    e.printStackTrace();
                                }
                            } else if (prepareMsgList.size() < 2) { // 准备阶段消息
                                prepareCdl.countDown();
                                System.out.println("(准备阶段)，" + whoAmI + "收到消息: " + msg);
                                prepareMsgList.add(msg);
                                // 发送提交阶段消息
                                if (prepareMsgList.size() == 2) {
                                    try {
                                        sendMsg(ATTACK_MSG, HAN_PORT);
                                        sendMsg(ATTACK_MSG, WEI_PORT);
                                        sendMsg(ATTACK_MSG, CHU_PORT);

                                        // 返回给客户端提交成功的消息
                                        new Thread(() -> {
                                            try {
                                                Thread.sleep(2000);
                                                sendMsg("commit成功", SUQIN_PORT);
                                            } catch (Exception e) {
                                                e.printStackTrace();
                                            }
                                        }).start();

                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }
                            } else if (commitPhaseMsgList.size() < 2) { // 提交阶段消息
                                try {
                                    prepareCdl.await();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                commitPhaseMsgList.add(msg);
                                System.out.println("(提交阶段)，" + whoAmI + "收到消息: " + msg);
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();

                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, ZHAO_NAME);
        zhao.start();

        // 副本
        Thread wei = new Thread(() -> {
            try {
                String whoAmI = WEI_NAME;
                ServerSocket serverSocket = new ServerSocket(WEI_PORT);
                allNodeReady.countDown();
                new Thread(() -> {
//                    int counter = 0;
                    // 预准备阶段消息
                    String prePrepareMsg = "";
                    // 收到的所有消息
                    List<String> allReceivedMsgList = new ArrayList<>();
                    // 准备阶段消息
                    List<String> preparePhaseMsgList = new ArrayList<>();
                    List<String> commitPhaseMsgList = new ArrayList<>();
                    while (true) {
//                        counter++;
                        try {
                            String msg = receiveMsg(serverSocket);
                            allReceivedMsgList.add(msg);
                            // 如果是与准备阶段消息的话
                            if (prePrepareMsg == "") {
                                System.out.println(whoAmI + "(预准备阶段)，收到消息：" + msg);
                                prePrepareMsg = msg;
                                prePrepareCdl.countDown();
                            } else if (preparePhaseMsgList.size() <= 0) {
                                prepareCdl.countDown();
                                preparePhaseMsgList.add(msg);
                                System.out.println("(准备阶段)，" + whoAmI + "收到消息：" + msg);
                                // 发送提交阶段消息
                                if (preparePhaseMsgList.size() > 0) {
                                    try {
                                        sendMsg(ATTACK_MSG, ZHAO_PORT);
                                        sendMsg(ATTACK_MSG, HAN_PORT);
                                        sendMsg(ATTACK_MSG, CHU_PORT);

                                        // 返回给客户端提交成功的消息
//                                        Thread.sleep(2000);
//                                        sendMsg("commit成功", SUQIN_PORT);
                                        new Thread(() -> {
                                            try {
                                                Thread.sleep(2000);
                                                sendMsg("commit成功", SUQIN_PORT);
                                            } catch (Exception e) {
                                                e.printStackTrace();
                                            }
                                        }).start();

                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }
                            } else if (commitPhaseMsgList.size() < 2) { // 提交阶段消息
                                try {
                                    prepareCdl.await();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                commitPhaseMsgList.add(msg);
                                System.out.println("(提交阶段)，" + whoAmI + "收到消息: " + msg);
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();

                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");

                //  准备阶段，给赵韩楚发消息
                prePrepareCdl.await();
                sendMsg(ATTACK_MSG, ZHAO_PORT);
                sendMsg(ATTACK_MSG, HAN_PORT);
                sendMsg(ATTACK_MSG, CHU_PORT);

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, WEI_NAME);
        wei.start();

        // 副本
        Thread han = new Thread(() -> {
            try {
                String whoAmI = HAN_NAME;
                ServerSocket serverSocket = new ServerSocket(HAN_PORT);
                allNodeReady.countDown();
                new Thread(() -> {
//                    int counter = 0;
                    // 预准备阶段消息
                    String prePrepareMsg = "";
                    // 收到的所有消息
                    List<String> allReceivedMsgList = new ArrayList<>();
                    // 准备阶段消息
                    List<String> preparePhaseMsgList = new ArrayList<>();
                    List<String> commitPhaseMsgList = new ArrayList<>();
                    while (true) {
//                        counter++;
                        try {
                            String msg = receiveMsg(serverSocket);
                            allReceivedMsgList.add(msg);
                            // 如果是与准备阶段消息的话
                            if (prePrepareMsg == "") {
                                System.out.println(whoAmI + "(预准备阶段)，收到消息：" + msg);
                                prePrepareMsg = msg;
                                prePrepareCdl.countDown();
                            } else if (preparePhaseMsgList.size() <= 0) {
                                prepareCdl.countDown();
                                preparePhaseMsgList.add(msg);
                                System.out.println("(准备阶段)，" + whoAmI + "收到消息：" + msg);
                                // 发送提交阶段消息
                                if (preparePhaseMsgList.size() > 0) {
                                    try {
                                        sendMsg(ATTACK_MSG, WEI_PORT);
                                        sendMsg(ATTACK_MSG, ZHAO_PORT);
                                        sendMsg(ATTACK_MSG, CHU_PORT);

                                        // 返回给客户端提交成功的消息
//                                        Thread.sleep(2000);
//                                        sendMsg("commit成功", SUQIN_PORT);
                                        new Thread(() -> {
                                            try {
                                                Thread.sleep(2000);
                                                sendMsg("commit成功", SUQIN_PORT);
                                            } catch (Exception e) {
                                                e.printStackTrace();
                                            }
                                        }).start();

                                    } catch (Exception e) {
                                        e.printStackTrace();
                                    }
                                }
                            } else if (commitPhaseMsgList.size() < 2) { // 提交阶段消息
                                try {
                                    prepareCdl.await();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                commitPhaseMsgList.add(msg);
                                System.out.println("(提交阶段)，" + whoAmI + "收到消息: " + msg);
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();

                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");
                //  准备阶段，给赵魏楚发消息
                prePrepareCdl.await();
                sendMsg(ATTACK_MSG, ZHAO_PORT);
                sendMsg(ATTACK_MSG, WEI_PORT);
                sendMsg(ATTACK_MSG, CHU_PORT);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }, HAN_NAME);
        han.start();

        // 副本（坏的）
        Thread chu = new Thread(() -> {
            try {
                String whoAmI = CHU_NAME;
                ServerSocket serverSocket = new ServerSocket(CHU_PORT);
                new Thread(() -> {
//                    int counter = 0;
                    // 预准备阶段消息
                    String prePrepareMsg = "";
                    // 收到的所有消息
                    List<String> allReceivedMsgList = new ArrayList<>();
                    // 准备阶段消息
                    List<String> preparePhaseMsgList = new ArrayList<>();
                    List<String> commitPhaseMsgList = new ArrayList<>();
                    while (true) {
//                        counter++;
                        try {
                            String msg = receiveMsg(serverSocket);
                            allReceivedMsgList.add(msg);
                            // 如果是与准备阶段消息的话
                            if (prePrepareMsg == "") {
                                System.out.println(whoAmI + "(预准备阶段)，收到消息：" + msg);
                                prePrepareMsg = msg;
                                prePrepareCdl.countDown();
                            } else if (preparePhaseMsgList.size() <= 1) {
                                prepareCdl.countDown();
                                preparePhaseMsgList.add(msg);
                                System.out.println("(准备阶段)，" + whoAmI + "收到消息：" + msg);
                            } else if (commitPhaseMsgList.size() < 3) { // 提交阶段消息
                                try {
                                    prepareCdl.await();
                                } catch (InterruptedException e) {
                                    e.printStackTrace();
                                }
                                commitPhaseMsgList.add(msg);
                                System.out.println("(提交阶段)，" + whoAmI + "收到消息: " + msg);
                            }
                        } catch (IOException e) {
                            e.printStackTrace();
                        }
                    }
                }).start();

                allNodeReady.countDown();
                allNodeReady.await();
                System.out.println(Thread.currentThread().getName() + " 准备好了！");

            } catch (Exception e) {
                e.printStackTrace();
            }
        }, CHU_NAME);
        chu.start();





        notExists.await();
    }

    public static String receiveMsg(ServerSocket serverA) throws IOException {
        Socket socketA = serverA.accept();
        InputStream is = socketA.getInputStream();
        InputStreamReader isr = new InputStreamReader(is);
        BufferedReader br = new BufferedReader(isr);
        while (true) {
            try {
                String s = br.readLine();
                if (s == null) {
                    break;
                    // 模拟第二轮发送消息，这里发送的是第一轮收到的消息
                }
                return s;
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        return "";
    }

    public static void sendMsg(String sendMsg, int port) throws Exception {
        Socket socket = new Socket();
        // 设置超时不生效，这里手动休眠
        Thread.sleep(200);
        socket.connect(new InetSocketAddress("127.0.0.1", port));
        OutputStream os = socket.getOutputStream();
        PrintWriter pw = new PrintWriter(os);
        pw.println(sendMsg);
        pw.flush();
        socket.close();
    }

}
